package com.camemax.utils;

import com.camemax.pojo.ItemViewCount;
import com.camemax.pojo.UserBehavior;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;

public class HotItemOperators {

    // 实现UserBehavior的累加计算
    public static class UserBehaviorAggregateFunctionImpl implements AggregateFunction<UserBehavior, Long, Long>{

        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(UserBehavior value, Long accumulator) {
            return accumulator + 1;
        }

        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return a + b;
        }
    }

    /*
    * 实现对UserBehavior的封装操作
    * */
    public static class ItemViewCountWindowFunctionImpl implements WindowFunction<Long, ItemViewCount, Long, TimeWindow>{
        @Override
        public void apply(Long inputItemId, TimeWindow window, Iterable<Long> input, Collector<ItemViewCount> out) {
            Long count = input.iterator().next();
            Long windowEnd = window.getEnd();
            out.collect(new ItemViewCount(inputItemId,windowEnd,count));
        }
    }

    /*
    * 实现数据收集，采用最底层的自定义分组处理函数KeyedProcessFunction
    * */
    public static class ItemViewCountKeyedProcessFunctionImpl extends KeyedProcessFunction<Long, ItemViewCount, String>{

        private int topNumber;

        public ItemViewCountKeyedProcessFunctionImpl(int topNumber){
            this.topNumber = topNumber;
        }

        ListState<ItemViewCount> itemViewCountListState ;

        @Override
        public void open(Configuration parameters) {
            itemViewCountListState = getRuntimeContext()
                    .getListState(new ListStateDescriptor<>("item-view-count-liststate", ItemViewCount.class));
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            ArrayList<ItemViewCount> itemViewCountArrayList = Lists.newArrayList(itemViewCountListState.get().iterator());
            itemViewCountArrayList.sort((itemViewCount1, itemViewCount2) -> itemViewCount2.getItemCount().intValue() - itemViewCount1.getItemCount().intValue());
            StringBuilder result = new StringBuilder();
            result.append(" ==================== \n");
            result.append(" 窗口结束时间: ").append(new Timestamp(timestamp - 1)).append("\n");
            for ( int i = 0 ; i < Math.min(itemViewCountArrayList.size(), topNumber) ; i ++){

                ItemViewCount currentItemView = itemViewCountArrayList.get(i);
                result.append(" Sale-Top.").append(i+1).append("\t").append("ItemId :").append(currentItemView.getItemId());
                result.append(" Sale-Count: ").append(currentItemView.getItemCount()).append("\n");
            }
            result.append(" ==================== \n\n");
            System.out.println(result);
            out.collect(result.toString());
        }

        @Override
        public void processElement(ItemViewCount value, Context ctx, Collector<String> out) throws Exception {
            itemViewCountListState.add(value);
            ctx.timerService().registerEventTimeTimer( value.getTimstamp() + 1 );
        }
    }
}
